package com.bbx.flink.demo.windowfuncation;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

@Slf4j
public class AggregateFuncationDemo {
    public static void main(String [] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.socketTextStream("192.168.10.131",10003)
                .map((MapFunction<String ,Integer> )value->Integer.parseInt(value))
                .keyBy(i->i%2)
                .timeWindow(Time.minutes(2))
                .aggregate((new AggregateFunction<Integer, Double, String>() {
                    private static final long serialVersionUID = 3906208733790618412L;

                    @Override
                    public Double createAccumulator() {
                        log.info("！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！创建累加器");
                        return 11.00;
                    }

                    @Override
                    public Double add(Integer value, Double accumulator) {
                        log.info("！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！数据相加");
                        return new Double(value)+ accumulator;
                    }

                    @Override
                    public String getResult(Double accumulator) {
                        log.info("！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！获取结果");
                        return String.valueOf(accumulator);
                    }

                    @Override
                    public Double merge(Double a, Double b) {
                        log.info("！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！数据合并");
                        return a+b;
                    }
                }))
                .print();

        env.execute();
    }
}
